package com.zlx.rulemk.demos;

import org.apache.commons.lang3.RandomUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * @Author: zhanglingxing
 * @CreateTime: 2022-07-07
 * @Description: 测试kafka log size 5M
 * @Version: 1.0
 */
public class TestKafkaLogSize5M {

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        Properties props = new Properties();
        props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "bigdata:9092");

        // key,value序列化（必须）：key.serializer，value.serializer
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        /**
         * 请求的大小，单位字节
         */
        props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 5242880);

        // 3. 创建kafka生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(props);

        StringBuilder sb = new StringBuilder();
        int length = 0;
        while(true){

            String s = "C";
            sb.append(s);
            length += utf8Length(s);
            if(length >= 5240000) {
                ProducerRecord<String, String> record = new ProducerRecord<>("test_topic",0,"1", sb.toString());
                kafkaProducer.send(record);
                kafkaProducer.flush();
                System.out.println("日志数据已发送：大小="+length);
                try {
                    Thread.sleep(RandomUtils.nextInt(1000, 5000));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                sb.setLength(0);
                length = 0;
            }
        }

    }

    public static int utf8Length(CharSequence cs) {
        return cs.codePoints()
                .map(cp -> cp<=0x7ff? cp<=0x7f? 1: 2: cp<=0xffff? 3: 4)
                .sum();
    }
}
